package com.hfzy.ihk.web.aliyuncall.support.disruptor;

import com.hfzy.ihk.web.aliyuncall.support.disruptor.base.DisruptorEvent;
import com.hfzy.ihk.web.aliyuncall.support.disruptor.base.DisruptorExecutorType;
import com.hfzy.ihk.web.aliyuncall.support.disruptor.base.DisruptorHandlerAdapter;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;

import java.text.MessageFormat;
import java.util.concurrent.Executor;

/**
 * Created by Administrator on 2016/5/17.
 * 未完成封装
 */
public class DisruptorFactoryBean implements FactoryBean<Disruptor>,InitializingBean,DisposableBean {

    private static final org.slf4j.Logger logger = LoggerFactory.getLogger(DisruptorFactoryBean.class);

    Disruptor disruptor = null;
    DisruptorUsingType disruptorUsingType = null;

    Executor executor = null;

    DisruptorExecutorType executorType = null;

    EventFactory<DisruptorEvent> eventFactory  = null;

    public int bufferSize = 1024;

    WaitStrategy waitStrategy = new BlockingWaitStrategy();

    ProducerType producerType = ProducerType.MULTI;

    DisruptorHandlerAdapter<DisruptorEvent> handlers[] = null;

    @Override
    public void afterPropertiesSet() throws Exception {

        if(disruptorUsingType == null)
            throw new RuntimeException("disruptorUsingType is require!");
        if(executorType == null)
            throw new RuntimeException("executorType is require!");
        if(eventFactory == null)
            throw new RuntimeException("eventFactory is require!");


       /* switch (disruptorUsingType){
            *//*disruptor用于呼入*//*
            case callIn://TODO 待改成 DefaultCallStateHandler/DefaultDynamicTransHandler/DefaultRecordDealHandler
                switch (executorType){
                    case SINGLE:
                    case FIXED:
                        executor = Executors.newFixedThreadPool(PublicConfig.DISRUPTOR_CALL_IN_THREAD_POOL_SIZE);
                        for(int i=0;i<PublicConfig.TELECOM_CALL_IN_HANDLER_SIZE;i++)
                            handlers[i] = new DefaultClickCallHandler();
                        break;
                    case DYNAMIC:
                        executor = Executors.newCachedThreadPool();
                        for(int i=0;i<PublicConfig.TELECOM_CALL_IN_HANDLER_SIZE;i++)
                            handlers[i] = new DefaultClickCallHandler();
                        break;
                }
                break;
            *//*disruptor用于呼出*//*
            case callOut:
                handlers = new DisruptorHandlerAdapter[PublicConfig.TELECOM_CALL_OUT_HANDLER_SIZE];
                switch (executorType){
                    *//*呼出肯定不使用单线程*//*
                    case SINGLE:
                    case FIXED:
                        executor = Executors.newFixedThreadPool(PublicConfig.DISRUPTOR_CALL_OUT_THREAD_POOL_SIZE);
                        for(int i=0;i<PublicConfig.TELECOM_CALL_OUT_HANDLER_SIZE;i++)
                            handlers[i] = (DisruptorHandlerAdapter<DisruptorEvent>)CloudCallInitializingBean.getApplicationContext().getBean("defaultClickCallHandler"); //new DefaultClickCallHandler();
                        break;
                    *//*暂时先使用这种*//*
                    case DYNAMIC:
                        executor = Executors.newCachedThreadPool();
                        for(int i=0;i<PublicConfig.TELECOM_CALL_OUT_HANDLER_SIZE;i++)
                            handlers[i] = (DisruptorHandlerAdapter<DisruptorEvent>)CloudCallInitializingBean.getApplicationContext().getBean("defaultClickCallHandler");//new DefaultClickCallHandler();
                        break;
                }
                break;
            *//*disruptor用于批量插入数据*//*
            case batchInsert:
                handlers = new DisruptorHandlerAdapter[1];
                switch (executorType){
                    case SINGLE:
                    case FIXED:
                    case DYNAMIC:
                        *//*只使用单线程处理*//*
                        executor = Executors.newSingleThreadExecutor();
                        handlers[0] = new DefaultBatchInsertAndWriteRedisHandler();
                        break;
                }
                break;
        }*/

        String format = MessageFormat.format("init disruptor:disruptorUsingType->[{0}],executorType->[{1}],bufferSize->[{2}],handlers size->[{3}]",
                disruptorUsingType.name(),
                executorType.name(),
                bufferSize,
                handlers.length);

        logger.debug(format);

        System.out.println(format);

        if(executor == null)
            throw new RuntimeException("executor is require!");

        disruptor = new Disruptor(eventFactory,
                bufferSize,
                executor,
                producerType,
                waitStrategy);

        if(handlers == null || handlers.length==0)
            throw new RuntimeException("handlers is require!");

//        for(DisruptorHandlerAdapter<DisruptorEvent> handlerAdapter : handlers)
//            disruptor.handleEventsWith(handlerAdapter);

        disruptor.handleEventsWith(handlers);

        disruptor.start();


        logger.debug("disruptor has started !");
    }

    @Override
    public Disruptor getObject() throws Exception {
        return disruptor;
    }

    @Override
    public Class<?> getObjectType() {
        return Disruptor.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    /**
     * Invoked by a BeanFactory on destruction of a singleton.
     *
     * @throws Exception in case of shutdown errors.
     *                   Exceptions will get logged but not rethrown to allow
     *                   other beans to release their resources too.
     */
    @Override
    public void destroy() throws Exception {
        disruptor.shutdown();
        executor = null;
        eventFactory = null;
        handlers = null;
    }

    public void setExecutor(Executor executor) {
        this.executor = executor;
    }

    public void setEventFactory(EventFactory<DisruptorEvent> eventFactory) {
        this.eventFactory = eventFactory;
    }

    public void setBufferSize(int bufferSize) {
        this.bufferSize = bufferSize;
    }

    public void setWaitStrategy(WaitStrategy waitStrategy) {
        this.waitStrategy = waitStrategy;
    }

    public void setProducerType(ProducerType producerType) {
        this.producerType = producerType;
    }

    public void setHandlers(DisruptorHandlerAdapter<DisruptorEvent>[] handlers) {
        this.handlers = handlers;
    }

    public void setDisruptorUsingType(DisruptorUsingType disruptorUsingType) {
        this.disruptorUsingType = disruptorUsingType;
    }

    public void setExecutorType(DisruptorExecutorType executorType) {
        this.executorType = executorType;
    }
}
